package shujia;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;

public class Demo3DianXin {
    /**
     * 1、建表dianxin
     * 2、将数据写入HBase
     * 3、需求：根据用户mdn，查询用户最新的3个位置记录（经纬度）
     */
    Connection conn;
    Admin admin;
    TableName dianxinTN;

    @Before
    public void createConn() throws IOException {
        // 1、创建一个配置文件
        Configuration conf = HBaseConfiguration.create();
        // 配置ZK的地址，通过ZK可以找到HBase
        conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

        // 2、创建连接
        conn = ConnectionFactory.createConnection(conf);
        // 3、创建Admin对象
        admin = conn.getAdmin();
        dianxinTN = TableName.valueOf("dianxin_split");

    }

    @Test
    /**
     * 建表 dianxin cf1
     */
    public void createDianXinTable() throws IOException {
        // 判断表存不存在，不存在则创建
        if (!admin.tableExists(dianxinTN)) {

            admin.createTable(new HTableDescriptor(dianxinTN)
                    .addFamily(new HColumnDescriptor("cf1")
                            .setMaxVersions(5)));
        } else {
            System.out.println("表已经存在");
        }
    }

    @Test
    /**
     * 插入数据
     */
    public void putAll() throws IOException {
        Table dianxin = conn.getTable(dianxinTN);

        BufferedReader br = new BufferedReader(new FileReader("data/DIANXIN.csv"));

        String line;

        int batchSize = 1000;
        ArrayList<Put> puts = new ArrayList<>();
        while ((line = br.readLine()) != null) {

            String[] splits = line.split(",");
            String mdn = splits[0];
            String start_time = splits[1];
            String lg = splits[4];
            String lat = splits[5];

            Put put = new Put(mdn.getBytes());

            put.addColumn("cf1".getBytes(), "lg".getBytes(), Long.parseLong(start_time), lg.getBytes());
            put.addColumn("cf1".getBytes(), "lat".getBytes(), Long.parseLong(start_time), lat.getBytes());

            puts.add(put);
            if (puts.size() == batchSize) {
                dianxin.put(puts);
                puts.clear();
            }

        }

        if (!puts.isEmpty()) {
            dianxin.put(puts);
        }

    }

    @Test
    /**
     * 根据mdn 获取该用户最新的3个位置记录
     * 22D3303E585F7E63AAFFF77E61A7A36E74BCE031
     */
    public void getPositionByMDN() throws IOException {
        Table dianxin = conn.getTable(dianxinTN);
        String mdn = "22D3303E585F7E63AAFFF77E61A7A36E74BCE031";
        Get get = new Get(mdn.getBytes());
        get.setMaxVersions(3);
        Result rs = dianxin.get(get);
        // 通过getValue这种方式只能获取到最新的记录
//        String lg = Bytes.toString(rs.getValue("cf1".getBytes(), "lg".getBytes()));
//        String lat = Bytes.toString(rs.getValue("cf1".getBytes(), "lat".getBytes()));
//
//        System.out.println(lg + "," + lat);
        // 定义两个ArrayList分别接收经纬度
        ArrayList<String> lgList = new ArrayList<>();
        ArrayList<String> latList = new ArrayList<>();

        // 对于多版本的数据 需要通过CellUtil的方式去取数据
        for (Cell cell : rs.listCells()) {
            String value = Bytes.toString(CellUtil.cloneValue(cell));
            String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));

            if ("lg".equals(columnName)) {
                lgList.add(value);
            } else if ("lat".equals(columnName)) {
                latList.add(value);
            }
        }
        for (int i = 0; i < 3; i++) {
            System.out.println(lgList.get(i)+","+latList.get(i));
        }

    }


    @After
    public void close() throws IOException {
        admin.close();
        conn.close();
    }
}
